package com.atguigu.sparkstreaming.demos

import org.apache.kafka.clients.consumer.ConsumerRecord
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.kafka010._
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe

/**
 * Created by Smexy on 2022/6/24
 */
object SimpleDemo {

  def main(args: Array[String]): Unit = {

    val streamingContext = new StreamingContext("local[*]", "simpledemo", Seconds(5))

    // 封装kafka的消费者参数
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "hadoop102:9092,hadoop103:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "test1",
      /*
          从主题的哪个位置开始消费。
            latest: 从主题的最新位置往后消费
            earliest: 从主题的最开始的位置往后消费
            none: 参考当前消费者组是否之前消费过这个主题，如果消费过，从上次记录的偏移量位置往后消费。
              如果配置的是latest，earliest仅仅在这个消费者组从未消费过这个主题是生效，一旦消费过后，即便配置的是latest，earliest，
              kafka也会设置为none，从消费过的offset位置往后消费。
       */
      "auto.offset.reset" -> "latest",
      // 是否允许kafka自动提交
      "enable.auto.commit" -> "true"
    )

    /*
        一般情况下，只写一个topic!
          在kafka中一个topic中存放的是一类数据。如果一个流，消费了两类数据，在这个DS后续的转换中，必须经过大量的判断，先判断当前这条数据
          是什么类型，再执行对应的转换，逻辑非常复杂。

          ds: TopicA ,TopicB。

          希望消费多个Topic，一般是创建多个DS，每个DS只消费一个Topic
     */
    val topics = Array("topicA")

    val ds: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](
      streamingContext,
      PreferConsistent,
      Subscribe[String, String](topics, kafkaParams)
    )

    //转换
    ds.map(record => record.value()).print()

    streamingContext.start()

    streamingContext.awaitTermination()

  }

}
